package com.bungarus.busi;

import com.bungarus.Busi;
import com.bungarus.codec.protobuf.MessageProtobuf;
import com.bungarus.config.BusiSpringConfig;
import com.bungarus.model.MessageBuilder;
import com.bungarus.model.Session;
import io.netty.channel.ChannelHandlerContext;
import org.apache.log4j.Logger;
import org.springframework.context.ApplicationContext;
import org.springframework.context.annotation.AnnotationConfigApplicationContext;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.StringRedisSerializer;
import org.springframework.util.StringUtils;

import java.util.*;

/**
 * Session management for authenticated user. It is a shared memory db supported by Redis that stores
 * all authenticated user session.
 * All methods in this type are asynchronous to ensure the large throughout of the server.
 *
 * The patter of session cached in redis likes:
 *   set  -> user:tenantid:userid dev1+app1 dev2+app2 dev3+app3
 *   hash -> tenantid:userid:dev1+app1 sessionid xxx ip_port xxx
 *
 * Created by tang on 2018/12/16.
 */
public class SessionManagement extends BusiProcessing {
    Logger logger = Logger.getLogger(SessionManagement.class);

    private static final String SESSION_TAG = "session_id";
    private static final String SESSION_IPPORT = "ip_port";
    private static final long SESSION_EXPIRE = 4 * 3600 * 1000;

    private static final String PREFIX_USER = "user:";
    private static final String APP_ALL = "ALL";
    private static final String DEV_ALL = "ALL";

    static final String REDIS_TEMPLATE = "redisStandaloneTemplate";
    static final String BUSI_SERVER = "busiServer";
    static final String VALUE_SEPERATOR = ",";

    static ApplicationContext springCtx = new AnnotationConfigApplicationContext(BusiSpringConfig.class);

    private RedisTemplate<String, Object> redisTemplate;

    private static SessionManagement sessionManagement = new SessionManagement();
    private SessionManagement(){
        init();
    }

    private void init() {
        this.redisTemplate = (RedisTemplate) springCtx.getBean(REDIS_TEMPLATE);
    }

    public static SessionManagement getInstanct() {
        return sessionManagement;
    }

    @Override
    public void doBusiness(Object o) throws BusiProcessingException {
        if(null == o || !(o instanceof Map)) {
            throw new IllegalArgumentException("Input argument must be instance of Map.");
        }
        Map map = (Map)o;
        ChannelHandlerContext ctx = (ChannelHandlerContext)map.get(SESSION_CONTEXT);
        MessageProtobuf.Message sessionMsg = (MessageProtobuf.Message)map.get(SESSION);

        BusiProcessing.busiTaskPool.execute(() -> {
            for(MessageProtobuf.Extension extension: sessionMsg.getHeader().getExtensionsList()){
                if (extension.getKey().equalsIgnoreCase(Session.CLIENT)){
                    String ip_port = extension.getValue();
                    save(
                            sessionMsg.getHeader().getFrom(),
                            String.valueOf(sessionMsg.getHeader().getTenantID()),
                            sessionMsg.getHeader().getApp(),
                            sessionMsg.getHeader().getDev(),
                            //user id which has prefix "user:"
                            sessionMsg.getHeader().getSessionId(),  // session id
                            ip_port);   // ip and port of access node to which the message will be sent

                    //Send ack signal to transfer node when session info has been saved successfully.
                    ctx.writeAndFlush(MessageBuilder.buildAckMessage(sessionMsg.getHeader().getId()));
                    break;
                }
            }

        });

    }

    /**
     * save session data
     * @param userId
     * @param tenantId
     * @param app
     * @param dev
     * @param sessionId
     * @param ipport
     */
    public void save(String userId, String tenantId, String app, String dev, String sessionId, String ipport) {
        logger.debug("save session:" + sessionId);
        //cache key pattern: user:{tenantId}:{userId}
        //cache value pattern: {userId},{sessionId},{ipport}
        //ipport format: {ip}:{port}
        String key = PREFIX_USER + tenantId + ":" + userId;
        //using redis's transaction function to fulfil the atomic operation
        String _app = StringUtils.isEmpty(app)?APP_ALL:app;
        String _dev = StringUtils.isEmpty(dev)?DEV_ALL:dev;

        StringRedisSerializer redisSerializer = new StringRedisSerializer();
        redisTemplate.executePipelined((RedisConnection redisConnection) -> {
            String hashKey = tenantId + ":" + userId + ":" + _app + ":" + _dev;

            redisConnection.multi();
            redisConnection.sAdd(redisSerializer.serialize(key), redisSerializer.serialize(_app + ":" + _dev));

            redisConnection.hSet(redisSerializer.serialize(hashKey),
                    redisSerializer.serialize(SESSION_TAG), redisSerializer.serialize(sessionId));
            redisConnection.hSet(redisSerializer.serialize(hashKey),
                    redisSerializer.serialize(SESSION_IPPORT), redisSerializer.serialize(ipport));

            redisConnection.expire(redisSerializer.serialize(key), SESSION_EXPIRE / 1000);
            redisConnection.expire(redisSerializer.serialize(hashKey), SESSION_EXPIRE / 1000);
            redisConnection.exec();

            return null;
        });
    }

    /**
     * get session data
     * @param userId
     * @param tenantId
     * @return
     */
    public Set<Object> retrieve(String userId, String tenantId) {
        Set<Object> devAppSet = redisTemplate.opsForSet().members(PREFIX_USER + tenantId + ":" + userId);
        StringRedisSerializer serializer = new StringRedisSerializer();
        Set<Object> result = new LinkedHashSet<>();
        List<Object> r = redisTemplate.executePipelined((RedisConnection redisConnection) -> {
            devAppSet.stream().forEach(devApp -> {
                //pattern: {dev}:{app}
                String strDevApp = (String)devApp;
                String hashKey = tenantId + ":" + userId + ":" + strDevApp;
                redisConnection.hMGet(serializer.serialize(hashKey), serializer.serialize(SESSION_TAG),
                        serializer.serialize(SESSION_IPPORT));
            });
            return null;
        });
        //pattern: {userId},{sessionId},{ipport}
        r.stream().forEach(l -> {
            //pattern: l instance of List<byte[]>
            List m = (List)l;
            result.add(userId + VALUE_SEPERATOR + m.get(0) + VALUE_SEPERATOR + m.get(1));
        });
        return result;
    }

    public List<Object> retrieveM(Set<Object> userIds, String tenantId) {
        logger.debug("retrieve multiple sessions");
        List<Object> results = new ArrayList<>();
        userIds.stream().forEach(userId -> {
            results.add(retrieve(userId.toString(), tenantId));
        });
        return results;
    }
}
